package com.foreveross.bsl.common.utils.balance.zookeeper;

import java.net.InetAddress;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * Dynamic member center. 集群管理
 * <p/>
 * The member maybe leave or dead dynamiclly.
 *
 * @author: heyaowen  Date: 2014-10-16 16:58:16
 */
public class GroupMemberCenter {

    private static final Log log = LogFactory.getLog(GroupMemberCenter.class);

    public static final String ESCAPE_PREFIX = "|||";
    private static final List<String> EMPTY_MEMBERS = new ArrayList<String>(0);

    private ZooKeeper zk;
    private String group = "";

    private String me = "";
    
	public void setZooKeeper(ZooKeeper zk) {
        this.zk = zk;
    }

	
    public boolean createAndSetGroup(String groupName) {
    	boolean result = this.createGroup(groupName);
        if (result) {
            this.setGroup(groupName);
        }
        return result;
    }

    public boolean createGroup(String groupName) {
        try {
            assert groupName != null;
            if (!groupName.startsWith("/")) {
                groupName = "/" + groupName;
            }
            Stat s = zk.exists(groupName, false);
            if (s == null) {
                zk.create(groupName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            log.error("create group error: " + groupName, e);
            return false;
        }
        return true;
    }

    public void setGroup(String groupName) {
        if (groupName != null && groupName.length() > 0) {
            if (!groupName.startsWith("/")) {
                groupName = "/" + groupName;
            }
            this.group = groupName;
        }
    }

    protected String buildName(String name) {
        return group + "/" + name;
    }

    public boolean joinGroupByDefine(String userdefine) {
        try {
            assert userdefine != null;
            assert userdefine.length() > 0;
            me = userdefine;
            if (me.contains("[host]")) {
                String host = InetAddress.getLocalHost().getHostAddress();
                me = me.replaceFirst("\\[host\\]", host);
            }
            //if contains "/", how to deal? --- maybe we need more format in future
            me = ESCAPE_PREFIX + URLEncoder.encode(me, "UTF-8");
            zk.create(this.buildName(me), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        } catch (Exception e) {
            log.error("join group error: " + me, e);
            return false;
        }
        return true;
    }

    public boolean joinGroup() {
        return joinGroup(null);
    }

    public boolean joinGroup(Integer port) {
        try {
            //use ipaddress as default, if you will use different ipaddress, you need joinGroup(yourip)
            me = InetAddress.getLocalHost().getHostAddress();
            return joinGroupByDefine(me + ":" + port);
        } catch (Exception e) {
            log.error("join group error", e);
            return false;
        }
    }

    public void leaveGroup() {
        try {
            zk.delete(this.buildName(me), 0);
        } catch (Exception e) {
            log.error("leave group error: " + me, e);
        }
    }

    public List<String> fetchGroupMembers() {
        return this.fetchGroupMembers(group, null);
    }

    public List<String> fetchGroupMembers(String groupName) {
        return this.fetchGroupMembers(groupName, null);
    }

    /** 获取节点信息，设置永久监听器
     * [fetchGroupMembers description]
     * @param  groupName [description]
     * @param  watcher   [description]
     * @return           [description]
     */
    public List<String> fetchGroupMembers(String groupName, Watcher watcher) {
		if (groupName != null && groupName.length() > 0) {
			if (!groupName.startsWith("/")) {
				groupName = "/" + groupName;
			}
		} else {
			return EMPTY_MEMBERS;
		}
		try {
			List<String> childlist;
			if (watcher == null) {
				childlist = zk.getChildren(groupName, false);
			} else {
				childlist = zk.getChildren(groupName, watcher);
			}
			List<String> lastresult = new ArrayList<String>();
			for (String item : childlist) {
				if (item.startsWith(ESCAPE_PREFIX)) {
					lastresult.add(URLDecoder.decode(item, "UTF-8").substring(3));
				} else {
					lastresult.add(item);
				}
			}
			return lastresult;
		} catch (Exception e) {
			log.error("fetch group members error", e);
			return EMPTY_MEMBERS;
		}
	}
}
